package com.lagou.kafka.demo.producer;


import com.lagou.kafka.demo.MyTransactional;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.IntegerSerializer;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

/**
 * @param
 * @author oyz
 * @version 1.0.4
 * @description: 关于kafka的事务操作
 * @date 2023/1/10 21:22
 */
public class MyTransactionalProducer {


    public static void main(String[] args) {
        KafkaProducer<String, String> producer = MyTransactional.getProducer();
        // 初始化事务
        producer.initTransactions();
        try{
            // 开启事务
            producer.beginTransaction();
            producer.send(new ProducerRecord<>("shenzhen","hello1","nice"));
            producer.send(new ProducerRecord<>("shenzhen","hello2","nice"));
            producer.send(new ProducerRecord<>("shenzhen","hello3","nice"));
            int i = 1/0;
            // 提交事务
            producer.commitTransaction();
        }catch (Exception e){
            // 回滚事务
            producer.abortTransaction();
        }finally {
            producer.close();
        }

    }


}
